.. _Совместное использование данных (Concurrency):

Совместное использование данных (Concurrency)
=============================================

.. index::
    concurrency

В предыдущей главе речь шла об обмене информацией между потоками посредством передачи сообщений
передачи сообщений. Как уже говорилось в этой главе, передача сообщений - это безопасный
метод параллелизма.
Другой метод предполагает чтение из одного потока и запись в
одни и те же данные. Например, поток-владелец может запустить рабочий поток с
адресом переменной bool, а рабочий может определить, завершать работу или нет.
читая текущее значение этой переменной. Другим примером может быть
когда владелец запускает несколько рабочих с адресом одной и той же переменной, так что
чтобы переменная была изменена более чем одним рабочим.
Одна из причин, по которой совместное использование данных небезопасно, - это условия гонки. Состояние гонки
возникает, когда более одного потока обращаются к одним и тем же изменяемым данным в
неконтролируемом порядке. Поскольку операционная система приостанавливает и запускает отдельные
потоков неопределенными способами, поведение программы, в которой есть условия гонки, непредсказуемо.
непредсказуемо.
Примеры, приведенные в этой главе, могут показаться упрощенными. Однако проблемы, которые они затрагивают
проявляются в реальных программах большего масштаба. Кроме того, хотя в этих примерах
используют модуль std.concurrency, концепции этой главы применимы и к модулю
core.thread.

Совместное использование данных не является автоматическим
----------------------------------------------------------

В отличие от большинства других языков программирования, в D данные не разделяются автоматически;
по умолчанию данные локальны в потоке. Хотя переменные на уровне модуля могут создавать
впечатление, что они доступны всем потокам, на самом деле каждый поток получает свою собственную
копию:

.. code-block:: d

    import std.stdio;
    import std.concurrency;
    import core.thread;

    int variable;
    void printInfo(string message)
    {
        writefln("%s: %s (@%s)", message, variable, &variable);
    }

    void worker()
    {
        variable = 42;
        printInfo("Before the worker is terminated");
    }

    void main()
    {
        spawn(&worker);
        thread_joinAll();
        printInfo("After the worker is terminated");
    }

Переменная, которая изменяется внутри worker(), - это не та переменная, которую видит
в main(). Этот факт можно заметить, распечатав как значения, так и
адреса переменных:

.. code-block:: d

.. code-block:: console

    Before the worker is terminated: 42 (@7F26C6711670)
    After the worker is terminated: 0 (@7F26C68127D0)

Поскольку каждый поток получает свою собственную копию данных, spawn() не позволяет передавать
ссылки на локальные переменные потока. Например, следующая программа, которая пытается
передать адрес переменной bool другому потоку, не может быть скомпилирована:

.. code-block:: d

    import std.concurrency;

    void worker(bool* isDone)
    {
        while (!(*isDone))
        {
            // ...
        }
    }

    void main()
    {
        bool isDone = false;
        spawn(&worker, &isDone); // ← compilation ERROR
        // ...
        // Hoping to signal the worker to terminate:
        isDone = true;
        // ...
    }

Статический assert внутри модуля std.concurrency предотвращает доступ к
mutabledata из другого потока:

.. code-block:: console

    src/phobos/std/concurrency.d(329): Error: static assert "Aliases to mutable thread-local data not allowed."

Адрес изменяемой переменной isDone не может быть передан между потоками.
Исключением из этого правила является переменная, которая определена как __gshared:

.. index::
    __gshared

.. code-block:: d

    __gshared int globallyShared;

Во всей программе существует только одна копия такой переменной, и все потоки
могут совместно использовать эту переменную. __gshared необходима при взаимодействии с библиотеками
таких языков, как C и C++, где разделение данных происходит автоматически по умолчанию. (???)

Cовместное использование изменяемых данных между потоками.
----------------------------------------------------------------------

Изменяемые переменные, которые должны быть общими, должны быть определены с помощью ключевого слова shared
ключевым словом shared:

.. code-block:: d


    import std.concurrency;

    void worker(shared(bool)* isDone)
    {
        while (*isDone)
        {
            // ...
        }
    }

    void main()
    {
        shared(bool) isDone = false;
        spawn(&worker, &isDone);
        // ...
        // Signalling the worker to terminate:
        isDone = true;
        // ...
    }

*Примечание:Предпочтите передачу сообщений для подачи сигналов потоку.*
С другой стороны, поскольку неизменяемые переменные не могут быть изменены, нет никакой
проблемы с их непосредственным совместным использованием. По этой причине immutable подразумевает shared:


.. code-block:: d

    import std.stdio;
    import std.concurrency;
    import core.thread;

    void worker(immutable(int)* data)
    {
        writeln("data: ", *data);
    }

    void main()
    {
        immutable(int) i = 42;
        spawn(&worker, &i); // ← compiles
        thread_joinAll();
    }

Вывод:

.. code-block:: console

    data: 42

Обратите внимание, что поскольку время жизни i определяется областью действия main(), важно, чтобы
чтобы main() не завершилась раньше рабочего потока. Вызов
core.thread.thread_joinAll выше, чтобы заставить поток дождаться завершения всех его дочерних потоков.

.. index::
    состояние гонки

Пример состояния гонки
----------------------

Корректность программы требует повышенного внимания, когда изменяемые данные
разделяются между потоками.
Чтобы увидеть пример состояния гонки, рассмотрим несколько потоков, совместно использующих
одну и ту же изменяемую переменную. Потоки в следующей программе получают
адреса как две переменные и меняют их значения местами большое количество раз:

.. code-block:: d

    import std.stdio;
    import std.concurrency;
    import core.thread;

    void swapper(shared(int)* first, shared(int)* second)
    {
        foreach (i; 0 .. 10_000)
        {
            int temp = *second;
            *second = *first;
            *first = temp;
        }
    }

    void main()
    {
        shared(int) i = 1;
        shared(int) j = 2;
        writefln("before: %s and %s", i, j);
        foreach (id; 0 .. 10)
        {
            spawn(&swapper, &i, &j);
        }
        // Wait for all threads to finish their tasks
        thread_joinAll();
        writefln("after : %s and %s", i, j);
    }

Хотя приведенная выше программа успешно компилируется, в большинстве случаев она будет работать
работать некорректно. Обратите внимание, что она запускает десять потоков, которые обращаются к одним и тем же двум
переменным i и j. В результате состояния гонки, в котором они находятся, они
непреднамеренно портят работу других потоков.
Также обратите внимание, что общее количество подмен составляет 10 раз по 10 тысяч. Поскольку это
это четное число, то естественно ожидать, что переменные в итоге будут иметь
значения 1 и 2 - их начальные значения:

.. code-block:: console

    before: 1 and 2
    after : 1 and 2 ← ожидаемый результат

Возможно, что программа действительно может дать такой результат, в большинстве случаев
чаще всего фактический результат будет одним из следующих:

.. code-block:: console

    before: 1 and 2
    after : 1 and 1 ← ошибочный результат

.. code-block:: console

before: 1 and 2
after : 2 and 2 ← ошибочный результат

Возможно, но крайне маловероятно, что в итоге может получиться даже "2 и 1".
тоже.
Причину некорректной работы программы можно объяснить следующим
следующим сценарием между двумя потоками, находящимися в состоянии гонки. По мере того как
операционная система приостанавливает и перезапускает потоки в неопределенное время, то
вероятен и следующий порядок выполнения операций двух потоков.
Рассмотрим состояние, когда i равно 1, а j равно 2. Хотя оба потока выполняют
одну и ту же функцию swapper(), помните, что локальная переменная temp является отдельной
для каждого потока и не зависит от других переменных temp других
потоков. Чтобы выделить эти отдельные переменные, ниже они будут переименованы в tempA и
tempB.
На диаграмме ниже показано, как трехстрочный код внутри цикла for может быть
выполняться каждым потоком с течением времени, сверху вниз, причем операция 1 является первой, а операция 6 - последней.
операция 1 - первая, а операция 6 - последняя. То, что i или j изменяется на
на каждом шаге, обозначается выделением этой переменной: (ToDo make a table)

.. code-block:: console

    Operation        Thread A                             Thread B
    ────────────────────────────────────────────────────────────────────────────
    1:   int temp = *second; (tempA==2)
    2:   *second = *first;   (i==1, j==1)
    (Assume that A is paused and B is started at this point)
    3:                                        int temp = *second; (tempB==1)
    4:                                        *second = *first;   (i==1, j==1)
    (Assume that B is paused and A is restarted at this point)
    5:   *first = temp;    (i==2, j==1)
    (Assume that A is paused and B is restarted at this point)
    6:                                        *first = temp;    (i==1, j==1)


Как видно, в конце предыдущего сценария и i, и j имеют значение
значение 1. Невозможно, чтобы после этого они могли иметь какое-либо другое значение.
Приведенный выше сценарий - лишь один пример, которого достаточно, чтобы объяснить неправильные
результаты работы программы. Очевидно, что условия гонки были бы гораздо более
в случае десяти потоков в этом примере.

Cинхронизация, чтобы избежать условий гонки
--------------------------------------------

Некорректное поведение программы, описанное выше, связано с тем, что более одного потока обращаются к
к одним и тем же изменяемым данным (и по крайней мере один из них изменяет их). Один из способов
избежать таких условий гонки - пометить общий код ключевым словом
synchronized. Программа будет корректно работать при следующем
изменении:

.. code-block:: d

    foreach (i; 0 .. 10_000)
    {
        synchronized
        {
            int temp =  * b;

            * b =  * a;

            * a = temp;
        }
    }

Вывод:

.. code-block:: console

    before: 1 and 2
    after : 1 and 2 ← ожидаемый результат

Эффект synchronized заключается в создании блокировки за кулисами и разрешении только
одному потоку удерживать эту блокировку в определенный момент времени. Только тот поток, который удерживает блокировку, может
выполняться, а остальные ждут, пока блокировка снова станет доступной, когда
когда выполняющий поток завершит свой синхронизированный блок. Поскольку один поток выполняет
синхронизированный код за раз, каждый поток теперь может безопасно поменять значения местами
прежде чем другой поток сделает то же самое. Состояние переменных i и j будет
всегда будет либо "1 и 2", либо "2 и 1" в конце обработки синхронизированного
блока.
*Примечание: ожидание блокировки является относительно дорогостоящей операцией для потока, что может
что может заметно замедлить выполнение программы. К счастью, в некоторых случаях корректность программы
корректность программы может быть обеспечена без использования асинхронизированного блока за счет использования
атомарных операций, которые будут описаны ниже.*
Когда необходимо синхронизировать более одного блока кода, можно
указать одну или несколько блокировок с помощью ключевого слова synchronized.
Рассмотрим пример этого в следующей программе, в которой есть два отдельных
блока кода, которые обращаются к одной и той же общей переменной. Программа вызывает две
функции с адресом одной и той же переменной, одна из которых инкрементирует, а
другая - декрементирует ее одинаковое количество раз:

.. code-block:: d

    void incrementer(shared(int)* value)
    {
        foreach (i; 0 .. count)
        {
            *value = *value + 1;
        }
    }

    void decrementer(shared(int)* value)
    {
        foreach (i; 0 .. count)
        {
            *value = *value - 1;
        }
    }

*Примечание:Если используются более короткие эквиваленты выражения выше (т.е. ++(*значение) и
--(*значение)), то компилятор предупреждает, что такие операции чтения-модификации-записи над
общими переменными являются устаревшими.*
К сожалению, пометить эти блоки отдельно синхронизированными недостаточно.
поскольку анонимные блокировки этих двух блоков будут независимы.
Таким образом, два блока кода все равно будут обращаться к одной и той же переменной одновременно:

.. code-block:: d

    import std.stdio;
    import std.concurrency;
    import core.thread;

    enum count = 1000;
    void incrementer(shared(int)* value)
    {
        foreach (i; 0 .. count)
        {
            synchronized
            { // ← This lock is different from the one below.
                *value = *value + 1;
            }
        }
    }

    void decrementer(shared(int)* value)
    {
        foreach (i; 0 .. count)
        {
            synchronized
            { // ← This lock is different from the one above.
                *value = *value - 1;
            }
        }
    }

    void main()
    {
        shared(int) number = 0;
        foreach (i; 0 .. 100)
        {
            spawn(&incrementer, &number);
            spawn(&decrementer, &number);
        }
        thread_joinAll();
        writeln("Final value: ", number);
    }

Поскольку существует равное количество потоков, которые инкрементируют и декрементируют одну и ту же переменную одинаковое количество раз
переменную одинаковое количество раз, можно ожидать, что конечное значение числа будет
нулю. Однако этого почти никогда не происходит:

.. code-block:: console

    Final value: -672

Чтобы несколько блоков использовали одну и ту же блокировку или блокировки, объекты блокировки должны быть
в синхронизированных круглых скобках:
*Примечание: эта возможность не поддерживается в dmd 2.098.1*

.. code-block:: d

    // Note: dmd 2.098.1 does not support this feature.
    synchronized (lock_object, another_lock_object, ...)


В D нет необходимости в специальном типе блокировки, потому что любой объект класса может быть использован
в качестве синхронизированной блокировки. Следующая программа определяет пустой класс с именем
Lock, чтобы использовать его объекты в качестве блокировок:

.. code-block:: d

    import std.stdio;
    import std.concurrency;
    import core.thread;

    enum count = 1000;
    class Lock
    {
    }

    void incrementer(shared(int)* value, shared(Lock) lock)
    {
        foreach (i; 0 .. count)
        {
            synchronized (lock)
            {
                *value = *value + 1;
            }
        }
    }

    void decrementer(shared(int)* value, shared(Lock) lock)
    {
        foreach (i; 0 .. count)
        {
            synchronized (lock)
            {
                *value = *value - 1;
            }
        }
    }

    void main()
    {
        shared(Lock) lock = new shared(Lock)();
        shared(int) number = 0;
        foreach (i; 0 .. 100)
        {
            spawn(&incrementer, &number, lock);
            spawn(&decrementer, &number, lock);
        }
        thread_joinAll();
        writeln("Final value: ", number);
    }

Поскольку на этот раз оба синхронизированных блока связаны одной и той же блокировкой,
в данный момент времени выполняется только один из них, и результат, как и ожидалось, равен нулю:

.. code-block:: console

    Final value: 0 ← correct result

Типы классов также могут быть определены как синхронизированные. Это означает, что все
нестатические функции-члены этого типа синхронизируются на данном объекте
данного класса:

.. code-block:: d

    synchronized class Class
    {
        void foo()
        {
            // ...
        }

        void bar()
        {
            // ...
        }
    }

Ниже приводится эквивалент определения класса, приведенного выше:

.. code-block:: d

    class Class
    {
        void foo()
        {
            synchronized (this)
            {
                // ...
            }
        }

        void bar()
        {
            synchronized (this)
            {
                // ...
            }
        }
    }

Когда блоки кода должны быть синхронизированы на более чем одном объекте, эти
должны быть указаны вместе. В противном случае возможно, что более чем один
поток может иметь заблокированные объекты, которые ожидают другие потоки, и в этом случае
программа может зайти в тупик.
Хорошо известный пример этой проблемы - функция, которая пытается перевести
деньги с одного банковского счета на другой. Чтобы эта функция работала корректно в
многопоточной среде, оба счета должны быть сначала заблокированы. Однако,
следующая попытка будет неверной:

.. code-block:: d

    void transferMoney(shared BankAccount from,
        shared BankAccount to)
    {
        synchronized (from)  // ошибка
        {
            {
                // ...
            }
        }
    }

Ошибку можно объяснить на примере, когда один поток пытается
перевести деньги со счета A на счет B, а другой поток пытается
перевести деньги в обратном направлении. Возможно, что каждый поток может
только что заблокировал свой соответствующий объект from, надеясь затем заблокировать свой объект to. Поскольку
объекты from соответствуют A и B в двух потоках соответственно, объекты
будут находиться в заблокированном состоянии в разных потоках, что делает невозможным для другого
невозможно, чтобы другой поток когда-либо заблокировал свой объект to. Такая ситуация называется тупиком.
Решение этой проблемы заключается в определении отношения упорядочивания между
объектами и блокировать их в этом порядке, что автоматически выполняется оператором
оператором synchronized. В D достаточно указать объекты в одном и том же
синхронизированном операторе, чтобы код избежал таких тупиков:
*Примечание: Эта возможность не поддерживается в dmd 2.098.1.*

.. code-block:: d

    void transferMoney(shared BankAccount from, shared BankAccount to)
    {
        // Note: dmd 2.098.1 does not support this feature.
        synchronized (from, to)
        { // ← correct
            // ...
        }
    }

.. index::
    shared static this

shared static this() для одиночной инициализации и shared static ~this() для одиночной финализации
--------------------------------------------------------------------------------------------------

Мы уже видели, что static this() можно использовать для инициализации модулей,
включая их переменные. Поскольку по умолчанию данные являются потоково-локальными, static this()
должна выполняться в каждом потоке, чтобы переменные уровня модуля были инициализированы для
всех потоков:

.. code-block:: d

    import std.stdio;
    import std.concurrency;
    import core.thread;

    static this()
    {
        writeln("executing static this()");
    }

    void worker()
    {
    }

    void main()
    {
        spawn(&worker);
        thread_joinAll();
    }

Приведенный выше статический блок this() будет выполнен один раз для главного потока и
один раз для рабочего потока:

.. code-block:: console

    executing static this()
    executing static this()

Это создаст проблемы для общих переменных модуля, поскольку инициализация переменной
переменной более одного раза будет неправильной, особенно при параллельном использовании из-за условий гонки
условий (??? todo) . (Это относится и к неизменяемым переменным, поскольку они
неявно разделяются). Решение заключается в использовании разделяемых статических блоков this(), которые
выполняются только один раз в программе:

.. code-block:: d

    int a; // thread-local
    immutable int b; // shared by all threads
    static this()
    {
        writeln("Initializing per-thread variable at ", &a);
        a = 42;
    }

    shared static this()
    {
        writeln("Initializing per-program variable at ", &b);
        b = 43;
    }

Вывод:

.. code-block:: console

    Initializing per-program variable at 6B0120 ← only once
    Initializing per-thread variable at 7FBDB36557D0
    Initializing per-thread variable at 7FBDB355467

Аналогично, общие static ~this() предназначены для финальных операций, которые должны выполняться
только один раз в программе.

.. index::
    атомарные операции

Атомарные операции
------------------

Другой способ гарантировать, что только один поток мутирует определенную переменную, - это
использование атомарных операций, функциональность которых обеспечивается
микропроцессором, компилятором или операционной системой.
Атомарные операции D находятся в модуле core.atomic. В этой главе мы рассмотрим только
две его функции в этой главе:

**atomicOp**

Эта функция применяет свой параметр шаблона к двум параметрам функции. Шаблонный параметр
должен быть бинарным оператором, например "+", "+=" и т. д.

.. code-block:: d

    import core.atomic;
    // ...
    atomicOp!"+="(*value, 1); // атомарная операция

Строка выше эквивалентна следующей строке, с той разницей, что операция
+= будет выполняться без прерывания другими потоками (т.е.
будет выполняться атомарно)

.. code-block:: d

    *value += 1; // не атомарная

Следовательно, если синхронизировать нужно только двоичную операцию,
то нет необходимости в синхронизированном блоке, который, как известно, работает медленно
из-за необходимости получения блокировки. Следующие эквиваленты функций
инкрементатор() и декрементатор(), использующие atomicOp, также корректны.
также корректны. Обратите внимание, что класс Lock также больше не нужен:

.. code-block:: d

    import core.atomic;

    //...
    void incrementer(shared(int)* value)
    {
        foreach (i; 0 .. count)
        {
            atomicOp!"+="(*value, 1);
        }
    }

    void decrementer(shared(int)* value)
    {
        foreach (i; 0 .. count)
        {
            atomicOp!"-="(*value, 1);
        }
    }

atomicOp можно использовать с другими бинарными операторами.

**cas**

Название этой функции - сокращение от "сравнить и поменять местами". Ее поведение
можно описать как мутацию переменной, если она все еще имеет известное значение. Для этого
используется при одновременном указании текущего и желаемого значений переменной
одновременно:

.. code-block:: d

    bool is_mutated = cas(address_of_variable, currentValue, newValue);

Тот факт, что значение переменной по-прежнему равно currentValue, когда работает cas()
является признаком того, что ни один другой поток не мутировал переменную с тех пор, как она
с момента последнего чтения переменной этим потоком. Если это так, то cas() присваивает переменной значение newValue
и возвращает true. С другой стороны, если значение переменной отличается от
currentValue, то cas() не изменяет переменную и возвращает false.
Следующие функции перечитывают текущее значение и вызывают cas() до тех пор, пока операция
пока операция не завершится успешно. Опять же, эти вызовы можно описать так: если значение переменной
переменной равно старому значению, замените его новым значением

.. code-block:: d

    void incrementer(shared(int)* value)
    {
        foreach (i; 0 .. count)
        {
            int currentValue;
            do
            {
                currentValue = *value;
            }
            while (!cas(value, currentValue, currentValue + 1));
        }
    }

    void decrementer(shared(int)* value)
    {
        foreach (i; 0 .. count)
        {
            int currentValue;
            do
            {
                currentValue = *value;
            }
            while (!cas(value, currentValue, currentValue - 1));
        }
    }


Приведенные выше функции корректно работают без синхронизированных блоков.
В большинстве случаев функции модуля core.atomic могут быть в несколько раз
быстрее, чем при использовании синхронизированных блоков. Я рекомендую обратить внимание на этот
модуль, если операции, требующие синхронизации, меньше блока кода.
Атомарные операции позволяют также использовать безблокировочные структуры данных, что выходит за рамки
этой книги.
Возможно, вы также захотите изучить пакет core.sync, который содержит
классические примитивы параллелизма в следующих модулях:
* core.sync.barrier
* core.sync.condition
* core.sync.config
* core.sync.exception
* core.sync.mutex
* core.sync.rwmutex
* core.sync.semaphore

Резюме
------

* Когда потоки не зависят от других потоков, предпочтите параллелизм. Рассматривайте
  конкурентное исполнение только тогда, когда потоки зависят от операций других потоков.
* Даже в этом случае отдавайте предпочтение параллелизму с передачей сообщений, о котором шла речь в
  предыдущей главе.
* Только общие данные могут быть общими; неизменяемые данные неявно являются общими.
* __gshared обеспечивает совместное использование данных, как в языках C и C++.
* Синхронизированный предназначен для предотвращения вмешательства других потоков, когда поток
  когда поток выполняет определенный фрагмент кода.
* Класс может быть определен как синхронизированный, чтобы только одна функция-член
  может выполняться на данном объекте в определенный момент времени. Другими словами, поток может
  выполнять функцию-член только в том случае, если ни один другой поток не выполняет функцию-член
  на том же объекте.
* статическая функция this() выполняется один раз для каждого потока; общая статическая функция this() выполняется один раз для всей программы.
  выполняется один раз для всей программы.
* Модуль core.atomic обеспечивает безопасный обмен данными, который может быть в несколько раз
  быстрее, чем при синхронизации.
* Пакет core.sync включает в себя множество других примитивов параллелизма
